package distributedbank.server;

import java.io.IOException;
import java.io.Serializable;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map.Entry;

import distributedbank.RequestThread;
import distributedbank.Wrapper;
import distributedbank.requests.SsRequest;
import distributedbank.responses.SsResponse;
import distributedbank.snapshot.Snapshots;

/**
 * A BankBranch object represents a branch of the bank with a branchId. The
 * branch contains several accounts. Most of the methods in a BankBranch object
 * must be thread-safe, since BankBranch objects are shared across multiple
 * Threads.
 */
public class BankBranch
{
    private static final int SS_REQ_ID_INCREMENT = 10;

    private Wrapper wrapper = new Wrapper();

    /**
     * local snapshot identifier
     */
    private int ssReqId;

    private int branchId;

    private HashMap<Integer, Account> accounts;

    private int incomingChannelsCount;

    // Bank Branch owns a local snapshots structure
    private Snapshots snapshots;

    public BankBranch( int branchId ) {
        this.branchId = branchId;
        this.ssReqId = branchId;
        accounts = new HashMap<Integer, Account>();
        snapshots = new Snapshots();
        incomingChannelsCount = Wrapper.incomingChannels( branchId ).size();
    }

    /**
     * Get snapshots in the bank server.
     */
    public Snapshots getSnapshots() {
        return snapshots;
    }

    /**
     * Returns the Account object with the given accountId. If no account with
     * the given account id exists, a new account with zero balance will be
     * returned.
     */
    public Account getAccount( int accountId ) {
        synchronized( accounts ) {
            if( accounts.containsKey( accountId ) ) {
                return accounts.get( accountId );
            } else {
                // the requested account doesn't exist, so we create a new
                // account with zero balance.
                Account emptyAccount = new Account();
                accounts.put( accountId, emptyAccount );
                return emptyAccount;
            }
        }
    }

    /**
     * @return a deep copy of the current states
     */
    public HashMap<Integer, Account> getState() {
        HashMap<Integer, Account> state = new HashMap<Integer, Account>();
        for( Entry<Integer, Account> pair : accounts.entrySet() ) {
            Account acct = new Account( pair.getValue().getBalance() );
            state.put( pair.getKey(), acct );
        }
        return state;
    }

    /**
     * Starts listening for connections on a socket on the specified port. For
     * every incoming connection, a new Server object is created to deal with
     * the message.
     */
    public void runServer() {
        try {
            ServerSocket serverSocket = new ServerSocket( Wrapper.basePort
                    + branchId );
            while( true ) {
                Socket clientSocket = serverSocket.accept();
                new RequestThread( this, clientSocket, wrapper ).start();
            }
        } catch( IOException e ) {
            System.out.println( e );
        }
    }

    /**
     * Receive a Snapshot Request. Call methods by this.snapshots
     * 
     * @param reqId
     *            Each SsRequest has a reqId.</br> For each snapshot, the id is
     *            unique.</br> The id's last digit shows from which bank
     *            branch's GUI that this snapshot request comes from.</br> </br>
     *            reqId == 0 shows the SsRequest comes from GUI.
     */
    public void recvSsReq( int reqId, int srcBranchId ) {
        if( !snapshots.getBlockedChannels().containsKey( reqId ) ) {
            snapshots.getBlockedChannels().put( reqId,
                    new LinkedList<Integer>() );
        }

        if( snapshots.getBlockedChannels().get( reqId ).contains( srcBranchId ) ) {
            // if already blocked, ignore the ss request
            return;
        }

        if( reqId == 0 ) {
            snapshots.storeState( ssReqId, getState() );
            broadcastSsReq( ssReqId );
            ssReqId += SS_REQ_ID_INCREMENT;
            return;
        }
        if( !snapshots.getChannels().containsKey( reqId ) ) {
            // State not recorded
            snapshots.storeState( reqId, getState() );
        }

        if( snapshots.getBlockedChannels().get( reqId ).contains( srcBranchId ) ) {
            // if already blocked, ignore the ss request
            return;
        }
        broadcastSsReq( reqId );
        snapshots.stopListening( reqId, srcBranchId );
        if( snapshots.getBlockedChannels().get( reqId ).size() == incomingChannelsCount ) {
            Serializable message = new SsResponse( reqId, snapshots
                    .getState( reqId ), snapshots.getChannel( reqId ) );
            wrapper.sendMessage( message, branchId, branchId + 4 );
            snapshots.cleanup( reqId );
        }
    }

    private void broadcastSsReq( int reqId ) {
        Serializable message = new SsRequest( reqId, branchId );
        for( int destBranch : Wrapper.whoNeighbors( branchId ) ) {
            if( destBranch < 5 ) {
                long sleeptime = 3000;
                try {
                    Thread.sleep( sleeptime );
                } catch( InterruptedException e ) {
                    e.printStackTrace();
                }
                wrapper.sendMessage( message, branchId, destBranch );
            }
        }
    }
}